package cn.edu.thu.laud.thrift;

import java.lang.reflect.Field;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import org.apache.cassandra.auth.Permission;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql.Attributes;
import org.apache.cassandra.cql.UpdateStatement;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.ColumnFamilyType;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.IColumn;
import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.RowMutation;
import org.apache.cassandra.db.SliceByNamesReadCommand;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.db.marshal.MarshalException;
import org.apache.cassandra.scheduler.IRequestScheduler;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.SocketSessionManagementService;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.CassandraServer;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.CounterColumn;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.ThriftValidation;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.edu.thu.laud.thrift.service.LaUDClientState;

/**
 * copy from CassandraServer. 主要是修改了clientState.将之换成了自己的clientstate（是LaudIfaceImpl的线程变量，是为了文件传输。）
 * @author hxd
 */
 public class LaUDCassandraServer extends CassandraServer {
	 // thread local state containing session information
    	//hxd: we replace this clientState by our LaUDClientState.
	private  ThreadLocal<ClientState> clientState=new ThreadLocal<ClientState>(){
		@Override
		protected ClientState initialValue() {
			logger.debug("create a new ClientStat:"+Thread.currentThread().getId());
			LaUDClientState laUDClientState=new LaUDClientState();
			laUDClientState.setSessionId("id"+Thread.currentThread().getId()+"t"+System.currentTimeMillis());
			LaUDDataServer.authorizedUsers.put(laUDClientState.getSessionId(), laUDClientState);
			return laUDClientState;
		}
	};
	
 
	private static Logger logger = LoggerFactory.getLogger(LaUDCassandraServer.class);
			  
    private final static List<ColumnOrSuperColumn> EMPTY_COLUMNS = Collections.emptyList();
    private final static List<Column> EMPTY_SUBCOLUMNS = Collections.emptyList();
    private final static List<CounterColumn> EMPTY_COUNTER_SUBCOLUMNS = Collections.emptyList();
    
    /*
     * RequestScheduler to perform the scheduling of incoming requests
     */
   //hxd:由于该字段在CassandraServer中是private的，无法继承，因此需要重新定义，并通过反射机制得到父类中的值，赋值到子类中。
    private  IRequestScheduler requestScheduler;
    /**
     * 与父类不同的是，对clientState进行赋值，requestScheduler使用父类的来赋值。
     * @param clientState
     * @throws RuntimeException
     */
	 
    public LaUDCassandraServer() throws RuntimeException{
    	super();
    	//this.clientState=clientState;
    	Field field;
		try {
			field = super.getClass().getSuperclass().getDeclaredField("requestScheduler");
			field.setAccessible(true);
			//Object object= field.get(this);
	    	this.requestScheduler=(IRequestScheduler) field.get(this);
		} catch (SecurityException e) {
			e.printStackTrace();
		} catch (NoSuchFieldException e) {
			e.printStackTrace();
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			e.printStackTrace();
		}

		if(this.requestScheduler==null){
			throw new RuntimeException("error.");
		}
    }
    /**
     * get the current user's clientstate. if not, create one and put it into a map.
     * hxd:差异点：创建的新的clientstate是laudClientState.
     */
	 
    public ClientState state()
    {
    	 SocketAddress remoteSocket = SocketSessionManagementService.remoteSocket.get();
        if (remoteSocket == null)
            return clientState.get();

        ClientState cState = SocketSessionManagementService.instance.get(remoteSocket);
        if (cState == null)
        {
       	//hxd: create a LaUDClientState instead of create a ClientState.
           cState = new LaUDClientState();
            SocketSessionManagementService.instance.put(remoteSocket, cState);
        }
        return cState;
    }
    
    
    /**
     * hxd: 由internal_get修改的。区别是，clientstate是传参得到的，而非state（）得到的。
     * <br>注意：为什么不用state()呢，因为internal_get方法是父类的private方法，其调用的state()是调用的父类的，而非子类重写的。。
     * @param state 
     * @param key
     * @param column_path
     * @param consistency_level
     * @return
     * @throws InvalidRequestException
     * @throws NotFoundException
     * @throws UnavailableException
     * @throws TimedOutException
     */
	 
    public Column file_internal_get(ClientState state,ByteBuffer key, ColumnPath column_path, ConsistencyLevel consistency_level)
    	    throws InvalidRequestException, NotFoundException, UnavailableException, TimedOutException
    	    {
    	        state.hasColumnFamilyAccess(column_path.column_family, Permission.READ);
    	        String keyspace = state.getKeyspace();
    	        CFMetaData metadata = ThriftValidation.validateColumnFamily(keyspace, column_path.column_family);
    	        ThriftValidation.validateColumnPath(metadata, column_path);
    	        ThriftValidation.validateConsistencyLevel(keyspace, consistency_level);

    	        QueryPath path = new QueryPath(column_path.column_family, column_path.column == null ? null : column_path.super_column);
    	        List<ByteBuffer> nameAsList = Arrays.asList(column_path.column == null ? column_path.super_column : column_path.column);
    	        ThriftValidation.validateKey(metadata, key);
    	        ReadCommand command = new SliceByNamesReadCommand(keyspace, key, path, nameAsList);
    	        Map<DecoratedKey, ColumnFamily> cfamilies = readColumnFamily(Arrays.asList(command), consistency_level);
    	        ColumnFamily cf = cfamilies.get(StorageService.getPartitioner().decorateKey(command.key));
    	        if (cf == null)
    	            throw new NotFoundException();
    	        List<ColumnOrSuperColumn> tcolumns = thriftifyColumnFamily(cf, command.queryPath.superColumnName != null, false);
    	        if (tcolumns.isEmpty())
    	            throw new NotFoundException();
    	        assert tcolumns.size() == 1;
    	        return tcolumns.get(0).column;
    	    }
    
    /**
     * hxd:没做修改，仅仅因为父类为private类型。
     * @param cf
     * @param subcolumnsOnly
     * @param reverseOrder
     * @return
     * @throws InvalidRequestException
     */
	 private List<ColumnOrSuperColumn> thriftifyColumnFamily(ColumnFamily cf, boolean subcolumnsOnly, boolean reverseOrder) throws InvalidRequestException
    {
        if (cf == null || cf.isEmpty())
            return EMPTY_COLUMNS;
        if (subcolumnsOnly)
        {
            IColumn column = cf.iterator().next();
            Collection<IColumn> subcolumns = column.getSubColumns();
            if (subcolumns == null || subcolumns.isEmpty())
                return EMPTY_COLUMNS;
            else
                return thriftifyColumns(subcolumns, reverseOrder);
        }
        if (cf.isSuper())
        {
           throw new InvalidRequestException("not support superColumn");
        }
        else
        {
            return thriftifyColumns(cf.getSortedColumns(), reverseOrder);
        }
    }
 /**
  * @author hxd
  * @param state
  * @param keyBuffer
  * @param columnFamily
  * @param column_name
  * @param column_value
  * @param attrs
  * @return
  * @throws InvalidRequestException
  * @throws UnavailableException
  * @throws TimedOutException
  */
    public boolean file_internal_insert(ClientState state, ByteBuffer keyBuffer,
			String columnFamily, ByteBuffer column_name, ByteBuffer column_value,
			Attributes attrs) throws InvalidRequestException, UnavailableException, TimedOutException {
    	state.hasColumnFamilyAccess(columnFamily, Permission.WRITE);
        Column column=new Column().setName(column_name).setValue(column_value).setTtl(attrs.getTimeToLive()).setTimestamp(attrs.getTimestamp()==null?state.getTimestamp():attrs.getTimestamp());
        return file_internal_insert(state, keyBuffer, new ColumnParent(columnFamily), column, attrs.getConsistencyLevel()==null?UpdateStatement.defaultConsistency:attrs.getConsistencyLevel());
	} 
///**
// * @author hxd     
// * @param state
// * @param key
// * @param columnFamily
// * @param columnName
// * @param columnValue
// * @param attrs
// * @return
// * @throws InvalidRequestException
// * @throws UnavailableException
// * @throws TimedOutException
// */
//    public boolean file_internal_insert(ClientState state, Term key,
//			String columnFamily, Term columnName, Term columnValue,
//			Attributes attrs) throws InvalidRequestException, UnavailableException, TimedOutException {
//    	state.hasColumnFamilyAccess(columnFamily, Permission.WRITE);
//    	CFMetaData metadata = ThriftValidation.validateColumnFamily(state.getKeyspace(), columnFamily, false);
//    	ByteBuffer keyBuffer=key.getByteBuffer(metadata.getKeyValidator());
//        ByteBuffer column_name=columnName.getByteBuffer(metadata.comparator);
//        ColumnDefinition cDefinition=metadata.getColumnDefinition(column_name);
//        ByteBuffer column_value=columnValue.getByteBuffer(cDefinition.getValidator());
//        
//        Column column=new Column().setName(column_name).setValue(column_value).setTtl(attrs.getTimeToLive()).setTimestamp(attrs.getTimestamp()==null?state.getTimestamp():attrs.getTimestamp());
//        return file_internal_insert(state, keyBuffer, new ColumnParent(columnFamily), column, attrs.getConsistencyLevel()==null?UpdateStatement.defaultConsistency:attrs.getConsistencyLevel());
//	}
    /**
     * 修改自父类的internal_insert。替换了其中的state()为传参得到。
     * @param state
     * @param key
     * @param column_parent
     * @param column
     * @param consistency_level
     * @return
     * @throws InvalidRequestException
     * @throws UnavailableException
     * @throws TimedOutException
     */
	 
    protected boolean file_internal_insert(ClientState state,ByteBuffer key, ColumnParent column_parent, Column column, ConsistencyLevel consistency_level) throws InvalidRequestException, UnavailableException, TimedOutException{
    	
        CFMetaData metadata = ThriftValidation.validateColumnFamily(state.getKeyspace(), column_parent.column_family, false);
        //ThriftValidation.validateKey(metadata, key);
        //ThriftValidation.validateColumnParent(metadata, column_parent);
        // SuperColumn field is usually optional, but not when we're inserting
        if (metadata.cfType == ColumnFamilyType.Super && column_parent.super_column == null)
        {
            throw new InvalidRequestException("missing mandatory super column name for super CF " + column_parent.column_family);
        }
        //ThriftValidation.validateColumnNames(metadata, column_parent, Arrays.asList(column.name));
       // ThriftValidation.validateColumnData(metadata, column, column_parent.super_column != null);

        RowMutation rm = new RowMutation(state.getKeyspace(), key);
        try
        {
            rm.add(new QueryPath(column_parent.column_family, column_parent.super_column, column.name), column.value, column.timestamp, column.ttl);
        }
        catch (MarshalException e)
        {
            throw new InvalidRequestException(e.getMessage());
        }
       return  file_doInsert( state,consistency_level, Arrays.asList(rm));
    }
    protected boolean file_doInsert(ClientState state, ConsistencyLevel consistency_level, List<RowMutation> mutations) throws UnavailableException, TimedOutException, InvalidRequestException{
    	ThriftValidation.validateConsistencyLevel(state.getKeyspace(), consistency_level);
        if (mutations.isEmpty())
            return true;
        try
        {
            file_schedule(state,DatabaseDescriptor.getRpcTimeout());
            try
            {
                StorageProxy.mutate(mutations, consistency_level);
            }
            finally
            {
            	 file_release();
            }
        }
        catch (TimeoutException e)
        {
            throw new TimedOutException();
        }
        return true;
    }
    /**
     * Schedule the current thread for access to the required services
     */
    protected void  file_schedule(ClientState state,long timeoutMS) throws TimeoutException
    {
        requestScheduler.queue(Thread.currentThread(), state.getSchedulingValue(), timeoutMS);
    }
    /**
     * Release count for the used up resources
     */
    protected void  file_release()
    {
        requestScheduler.release();
    }

	public ThreadLocal<ClientState> getThreadLocalClientStateField() {
		return clientState;
	}

	public void setThreadLocalClientStateField(ThreadLocal<ClientState> clientState) {
		this.clientState = clientState;
	}

	
	
}
